package com.github.rpc4j.common.network;

import static org.jboss.netty.channel.Channels.pipeline;

import java.util.ArrayList;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;
import java.util.concurrent.Executors;
import java.util.concurrent.atomic.AtomicInteger;

import org.jboss.netty.bootstrap.ClientBootstrap;
import org.jboss.netty.channel.ChannelPipeline;
import org.jboss.netty.channel.ChannelPipelineFactory;
import org.jboss.netty.channel.socket.nio.NioClientSocketChannelFactory;
import org.jboss.netty.handler.codec.serialization.ObjectDecoder;
import org.jboss.netty.handler.codec.serialization.ObjectEncoder;

import com.github.rpc4j.common.network.handler.NettyRpc4jClientHandler;

public class NettyRpcConnectionFactory implements RpcConnectionFactory<NettyRpcConnection>{
	
	private ClientBootstrap bootstrap = null;
	
	private NettyRpc4jClientHandler handler = new NettyRpc4jClientHandler();
	
	private final static AtomicInteger CONNECTION_ID = new AtomicInteger(0);
	
	private ConcurrentHashMap<String, LinkedList<NettyRpcConnection>> connectionPoolMap = new ConcurrentHashMap<String, LinkedList<NettyRpcConnection>>();
	
	
	public void init() {
		bootstrap = new ClientBootstrap(new NioClientSocketChannelFactory(Executors.newCachedThreadPool(), Executors.newCachedThreadPool(),50));
        bootstrap.setPipelineFactory(new ChannelPipelineFactory() {
			public ChannelPipeline getPipeline() throws Exception {
				ChannelPipeline pipeline = pipeline();
				pipeline.addLast("en", new ObjectEncoder());
				pipeline.addLast("de", new ObjectDecoder());
				pipeline.addLast("handler", handler);
				return pipeline;
			}
		});
   }
	
	public synchronized NettyRpcConnection getConnection(String serviceIp){
		LinkedList<NettyRpcConnection> poolList =   connectionPoolMap.get(serviceIp);
		if(poolList==null){
			poolList =  new LinkedList<NettyRpcConnection>();
			connectionPoolMap.put(serviceIp, poolList);
		}
		if(!poolList.isEmpty())
				return poolList.removeFirst();
		NettyRpcConnection connection = new NettyRpcConnection(CONNECTION_ID.getAndIncrement(),bootstrap,handler,serviceIp);
		connection.init();
		return connection;
	}
	
	public synchronized void returnConnection(NettyRpcConnection connection){
		if(connection==null)
			return;
		LinkedList<NettyRpcConnection> connectionList = connectionPoolMap.get(connection.getServiceIpStr());
		connectionList.addLast(connection);
	}
}
